package de.mmis.core.publishsubscribe.filter;

import java.util.HashMap;
import java.util.Map;

import de.mmis.core.base.DeserializableAnnotations.DeserializationMethod;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;

/**
 * An Atomic Filter compares {@link FilterValue}s
 * 
 * @author Martin Nyolt
 * 
 */
public class AtomicFilter extends Filter {

	private final FilterOperator op;
	private FilterValue value1, value2;
	private String ref;

	/**
	 * Empty Filter
	 */
	public AtomicFilter() {
		op = null;
	}

	/**
	 * creates a Filter that checks if the key is present in the Event
	 */
	public AtomicFilter(String key) {
		op = null;
		this.ref = key;
	}

	/**
	 * creates a Filter that compares two values.
	 * 
	 * @param op
	 *            The {@link FilterOperator} used to compare these two values.
	 * @param value1
	 * @param value2
	 */
	public AtomicFilter(FilterOperator op, FilterValue value1,
			FilterValue value2) {
		this.op = op;
		this.value1 = value1;
		this.value2 = value2;
	}

	/**
	 * @return the op
	 */
	public FilterOperator getOp() {
		return op;
	}

	/**
	 * @return the value1
	 */
	public FilterValue getValue1() {
		return value1;
	}

	/**
	 * @return the value2
	 */
	public FilterValue getValue2() {
		return value2;
	}

	/**
	 * @return the ref
	 */
	public String getRef() {
		return ref;
	}

	private static enum Type {
		Byte, Short, Int, Long, Float, Double, Boolean, Char, String, Symbol,
		Tree;
	}

	/**
	 * returns the {@link Type} of the {@link Tree}. If an expression has
	 * multiple numerical types, the smallest is choosen; regarding to the
	 * following order:
	 * 
	 * Byte < Short < Int < Long < Float < Double
	 */
	private Type getType(Tree e1) {
		Type result = null;

		if (e1.isBoolean())
			result = Type.Boolean;
		else if (e1.isByte())
			result = Type.Byte;
		else if (e1.isShort())
			result = Type.Short;
		else if (e1.isInt())
			result = Type.Int;
		else if (e1.isLong())
			result = Type.Long;
		else if (e1.isFloat())
			result = Type.Float;
		else if (e1.isDouble())
			result = Type.Double;
		else if (e1.isChar())
			result = Type.Char;
		else if (e1.isString() && !new LeafNode("null").equals(e1))
			result = Type.String;
		else if (e1.isSymbol())
			result = Type.Symbol;
		else
			result = Type.Tree;

		return result;
	}

	/**
	 * 
	 * @param e1
	 * @param e2
	 * 
	 * @return the primitive Java type both expression can be cast to compre
	 *         them, or <code>null</code> if they don't have a matching type
	 */
	private Type getBestType(Tree e1, Tree e2) {
		Type t1 = getType(e1);
		Type t2 = getType(e2);

		if (t1 == t2)
			return t1;

		// they are not the same type
		// if one of this types a not a numerical value, they are not compatible
		if (t1 == Type.Boolean || t1 == Type.Char || t1 == Type.String
				|| t1 == Type.Symbol || t1 == Type.Tree)
			return null;
		if (t2 == Type.Boolean || t2 == Type.Char || t2 == Type.String
				|| t2 == Type.Symbol || t2 == Type.Tree)
			return null;

		// both are numerical - always return the greater type

		int o1 = t1.ordinal();
		int o2 = t2.ordinal();
		int max = Math.max(o1, o2);

		return Type.values()[max];
	}

	@Override
	public boolean matches(Map<String, Tree> e,
			HashMap<String, Map<String, Tree>> worldstate)
			throws FilterException {
		if (op == null)
			return ref == null || e.containsKey(ref);

		Tree e1 = value1.getValue(e, worldstate);
		Tree e2 = value2.getValue(e, worldstate);

		if (e1 == null || e2 == null)
			return false;

		if (op == FilterOperator.IDENTICAL)
			return e1.equals(e2);

		if (op == FilterOperator.NOT_IDENTICAL)
			return !e1.equals(e2);

		if (op == FilterOperator.MATCH)
			return e1.matchPattern(e2);

		if (op == FilterOperator.NOT_MATCH)
			return !e1.matchPattern(e2);

		Type t = getBestType(e1, e2);

		if (t == null)
			throw new FilterException("Type " + getType(e1)
					+ " of first argument [" + value1 + ", value " + e1
					+ "] is not compatible with Type " + getType(e2)
					+ " of second argument [" + value2 + ", value " + e2 + "]");

		try {
			switch (t) {
			case Tree:
				return e1.equals(e2);
			case Boolean:
				boolean b1 = e1.getAsBoolean();
				boolean b2 = e2.getAsBoolean();
				return evaluate_boolean(b1, b2);
			case Char:
				char c1 = e1.getAsChar();
				char c2 = e2.getAsChar();
				return evaluate_char(c1, c2);
			case String:
				String s1 = e1.getAsString();
				String s2 = e2.getAsString();
				return evaluate_String(s1, s2);
			case Symbol:
				String sy1 = e1.getAsSymbol();
				String sy2 = e2.getAsSymbol();
				return evaluate_symbol(sy1, sy2);
			case Byte:
				byte by1 = e1.getAsByte();
				byte by2 = e2.getAsByte();
				return evaluate_byte(by1, by2);
			case Short:
				short sh1 = e1.getAsShort();
				short sh2 = e2.getAsShort();
				return evaluate_short(sh1, sh2);
			case Int:
				int i1 = e1.getAsInt();
				int i2 = e2.getAsInt();
				return evaluate_int(i1, i2);
			case Long:
				long l1 = e1.getAsLong();
				long l2 = e2.getAsLong();
				return evaluate_long(l1, l2);
			case Float:
				float f1 = e1.getAsFloat();
				float f2 = e2.getAsFloat();
				return evaluate_float(f1, f2);
			case Double:
				double d1 = e1.getAsDouble();
				double d2 = e2.getAsDouble();
				return evaluate_double(d1, d2);
			}
		} catch (DeserializableException de) {
			// should not happen
			throw new FilterException("Unexpected deserialize exception ", de);
		}

		return false;
	}

	private boolean evaluate_char(char c1, char c2) throws FilterException {
		switch (op) {
		case EQUAL:
			return c1 == c2;
		case NOT_EQUAL:
			return c1 != c2;
		case GREATER:
			return c1 > c2;
		case GREATER_EQUAL:
			return c1 >= c2;
		case LESS:
			return c1 < c2;
		case LESS_EQUAL:
			return c1 <= c2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_boolean(boolean b1, boolean b2)
			throws FilterException {
		switch (op) {
		case EQUAL:
			return b1 == b2;
		case NOT_EQUAL:
			return b1 != b2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_String(String s1, String s2)
			throws FilterException {
		switch (op) {
		case EQUAL:
			return s1.equals(s2);
		case NOT_EQUAL:
			return !s1.equals(s2);
		case GREATER:
			return s1.compareTo(s2) > 0;
		case GREATER_EQUAL:
			return s1.compareTo(s2) >= 0;
		case LESS:
			return s1.compareTo(s2) < 0;
		case LESS_EQUAL:
			return s1.compareTo(s2) <= 0;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_symbol(String s1, String s2)
			throws FilterException {
		switch (op) {
		case EQUAL:
			return s1.equals(s2);
		case NOT_EQUAL:
			return !s1.equals(s2);
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_byte(byte v1, byte v2) throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_short(short v1, short v2) throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_int(int v1, int v2) throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_long(long v1, long v2) throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_float(float v1, float v2) throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	private boolean evaluate_double(double v1, double v2)
			throws FilterException {
		switch (op) {
		case EQUAL:
			return v1 == v2;
		case NOT_EQUAL:
			return v1 != v2;
		case GREATER:
			return v1 > v2;
		case GREATER_EQUAL:
			return v1 >= v2;
		case LESS:
			return v1 < v2;
		case LESS_EQUAL:
			return v1 <= v2;
		default:
			throw new FilterException(
					"Boolean operands are not applicable for operator " + op);
		}
	}

	@Override
	public Tree[] serialize(boolean explicit, Map<Object, String> refmap,
			ClassLoader classLoader) {
		InnerNode result = new InnerNode();

		if (op == null && ref != null)
			result.add(Tree.fromObject(ref, explicit, refmap, classLoader));
		else if (op != null) {
			result.add(Tree.fromObject(op, explicit, refmap, classLoader));
			result.add(Tree.fromObject(value1, explicit, refmap, classLoader));
			result.add(Tree.fromObject(value2, explicit, refmap, classLoader));
		}

		return new Tree[] { result };
	}

	@Override
	public String getTypeIdentifier() {
		return "publishsubscribe/atomic-filter";
	}

	@DeserializationMethod
	public static AtomicFilter deserialize(Tree[] exps, ClassLoader cl,
			String id, Map<String, Object> idmap)
			throws DeserializableException {
		if (exps.length != 1)
			throw new DeserializableException(null,
					"Accept only one expression");

		Tree e = exps[0];
		if (!(e instanceof InnerNode))
			throw new DeserializableException(e, "Expression must be a struct");

		InnerNode s = (InnerNode) e;

		if (s.getNumberOfSubTrees() == 0) {
			return new AtomicFilter();
		}

		if (s.getNumberOfSubTrees() == 1) {
			String ref = s.getSubTree(0).getAsString();
			return new AtomicFilter(ref);
		}

		if (s.getNumberOfSubTrees() == 3) {
			FilterOperator op = s.getSubTree(0).getAs(FilterOperator.class);
			FilterValue v1 = s.getSubTree(1).getAs(FilterValue.class);
			FilterValue v2 = s.getSubTree(2).getAs(FilterValue.class);
			return new AtomicFilter(op, v1, v2);
		}

		throw new DeserializableException(s,
				"Struct must have 0, 1 or 3 sub-expressions");
	}

	@Override
	public boolean equals(Object obj) {
		if (obj == null)
			return false;
		if (!(obj instanceof AtomicFilter))
			return false;

		AtomicFilter o = (AtomicFilter) obj;

		if (o.op == null) {
			if (op != null)
				return false;
			return o.ref.equals(ref);
		}

		return o.op.equals(op) && o.value1.equals(value1)
				&& o.value2.equals(value2);
	}

	@Override
	public int hashCode() {
		if (op == null)
			return ref.hashCode();

		return op.hashCode() + value1.hashCode() + value2.hashCode();
	}

	@Override
	public String toString() {
		if (op == null) {
			if (ref != null)
				return "EXISTS " + ref;
			return "EMPTY";
		}

		return value1 + " " + op + " " + value2;
	}

}
